package com.yyy.tcc.service.compensation;

import com.yyy.tcc.service.AccountService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.stereotype.Service;
import org.springframework.beans.factory.annotation.Autowired;

@Service
public class KafkaCompensationService {
    private KafkaTemplate<String, String> kafkaTemplate;
    private AccountService accountService;
    private MessageListener<String, String> messageListener;

    @Autowired
    public KafkaCompensationService(KafkaTemplate<String, String> kafkaTemplate,
                                  AccountService accountService) {
        this.kafkaTemplate = kafkaTemplate;
        this.accountService = accountService;
    }

    public KafkaCompensationService() {
    }

    public void setKafkaTemplate(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void setAccountService(AccountService accountService) {
        this.accountService = accountService;
    }

    public void setMessageListener(MessageListener<String, String> messageListener) {
        this.messageListener = messageListener;
    }

    public void handleCompensationMessage(ConsumerRecord<String, String> record) {
        try {
            String transactionId = record.key();
            accountService.cancel(transactionId);
        } catch (Exception e) {
            throw new RuntimeException("处理补偿消息失败", e);
        }
    }

    /**
     * 关闭Kafka服务
     */
    public void shutdown() {
        if (kafkaTemplate != null) {
            try {
                kafkaTemplate.destroy();
            } catch (Exception e) {
                throw new RuntimeException("关闭Kafka服务失败", e);
            }
        }
    }
}
